<?php
namespace app\common\logic;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\facade\Config;

class Amqp extends Base{

    private $host;
    private $port;
    private $virtualHost;
    private $user;
    private $password;

    function __construct() {
        $config = Config::pull('amqp');
        $this->host         = $config['host'];
        $this->port         = $config['port'];
        $this->virtualHost  = $config['virtualHost'];
        $this->user  = $config['user'];
        $this->password  = $config['password'];
    }

    public function getConnection() {
        return new AMQPStreamConnection($this->host, $this->port, $this->user, $this->password, $this->virtualHost);
    }

    /**
     * @desc 接收消息队列
     * @param string $queueName 队列名称
     * @throws \ErrorException
     */
    public function receive($queueName) {
        $connection = $this->getConnection();
        $channel = $connection->channel();

        $channel->queue_declare($queueName, false, false, false, false);

        //echo " [*] Waiting for messages. To exit press CTRL+C\n";

        $callback = function ($msg) {
            $result = json_decode($msg->body, true);
            echo ' [x] Received ', $msg->body, "\n";
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };

        //消费者设置手动ack，在声明no_ack参数时指定false
        $channel->basic_consume($queueName, '', false, true, false, false, $callback);

        while (count($channel->callbacks)) {
            $channel->wait();
        }

        $channel->close();
        $connection->close();
    }



    /**
     * @desc  发送消息队列
     * @param string $queueName 队列名称
     * @param string $msg 发送消息
     * @return bool
     * @throws \Exception
     */
    public function send($queueName, $msg) {
        // 创建连接
        $connection = $this->getConnection();

        // 创建channel，多个channel可以共用连接
        $channel = $connection->channel();


        // 创建交换机以及队列（如果已经存在，不需要重新再次创建并且绑定）

        // 创建直连的交换机
        //$channel->exchange_declare('amq.direct', 'direct', false, false, false);

        // 队列创建申明 第三个参数为true时，表示队列持久化,需要和客户端一致
        $channel->queue_declare($queueName, false, true, false, false);

        // 交换机跟队列的绑定，
        //$channel->queue_bind($queueName, 'amq.direct', $queueName);

        $body = new AMQPMessage($msg);

        $channel->basic_publish($body, '', $queueName);

        $channel->close();
        $connection->close();
    }

    /**
     * @desc 发送队列消息
     * @param $queueName
     * @param $msg
     * @throws \Exception
     */
    public function sendWithProp($queueName, $msg) {
        $connection = $this->getConnection();

        $channel = $connection->channel();

        $channel->queue_declare('queue', false, false, false, false);

        $amqpTable = new AMQPTable(["delay"=>"1000"]);

        $body = new AMQPMessage($msg, ['application_headers'=>$amqpTable,'content_type' => 'text/plain', 'delivery_mode' => 2]); //生成消息

        $channel->basic_publish($body, '', $queueName);

        $channel->close();
        $connection->close();
    }
}